package com.intct.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.common.Constant;
import com.intct.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author
 * @date
 *
 **/
public class DwdVehicleMainApp {

    public static void main(String[] args) throws Exception {
        // 1.基本环境准备
        //1.1 指定流处理环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2001);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        //1.2 设置并行度
        env.setParallelism(1);

        // 启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 2.从kafka的主题中读取业务数据
        //2.1 声明消费的主题以及消费者组
        //2.2 创建消费者对象
        String ckAndGroupId = "dwd_vehicle_group_id";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(Constant.KAFAK_TOPIC_ODS_VEHICLE, ckAndGroupId);
        //2.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka_Source");

        // 3.对流中的数据类型进行转换
        SingleOutputStreamOperator<String> monitorDS = kafkaStrDS.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String jsonStr, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                try {
                    // 对流中的数据类型进行转换 jsonStr-> jsonObject
                    JSONObject jsonObject = JSON.parseObject(jsonStr);

                    // 获取数据
                    String afterObj = jsonObject.getString("after");

                    // 写出数据
                    out.collect(afterObj);

                } catch (Exception e) {
                    throw new RuntimeException("数据处理失败，非正常JSON数据格式！");
                }
            }
        });

        // 4. 过滤掉不在线车辆
        SingleOutputStreamOperator<String> filterDS = monitorDS.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String jsonStr) throws Exception {
                // 对流中的数据类型进行转换 jsonStr-> jsonObject
                JSONObject jsonObj = JSON.parseObject(jsonStr);

                // 车辆是否在线标识
                String type = jsonObj.getString("type");

                // 只有在线车辆保留
                return "0".equals(type);
            }
        });

        // 5. 脱敏 时间处理

        // 6. 写出数据到DWD Kafka
        filterDS.sinkTo(KafkaUtil.getKafkaSink(Constant.KAFAK_TOPIC_DWD_VEHICLE, "v_id"));

        // 7. 执行作业
        env.execute();
    }
}
